package com.demo.webflux.simple2;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import reactor.core.publisher.Flux;
import reactor.core.publisher.ReplayProcessor;

@RestController
public class MessageController {

	final ReplayProcessor<MessageResponse> messageStream = ReplayProcessor.create(50); // 使我们能缓存预定义数量的元素并将最新元素重播给每个新订阅者

	public MessageController(ChatService<MessageResponse> chatService) { // 构造函数声明。在构造函数中，我们构建一个处理流程，他合并来自gitter的最新消息流
		Flux.mergeSequential(chatService.getMessageAfter(null).flatMapIterable(Function.identity()), // 在空id的情况下，gitter返回30条最新消息。处理流程还近实时的监听新消息流。
				chatService.getMessageStream())
		.map(x -> x) // 然后所有的消息都映射到视图模型
		.subscribe(messageStream); // 并且流由ReplayProcessor立即进行订阅。这意味着一旦构造了MessageController bean，我们就连接到gitter服务，缓存最新消息并开始监听更新。
		// 请注意，mergeSequential虽然同时订阅两个流，但仅在第一个流完成时才开始从第二个流发送消息。由于第一个流是有限的，我们接收最新的消息并开始从getMessageStream Flux发送排队的消息
	}

    @GetMapping("/message")
	public Flux<MessageResponse> stream(){
    	// 我们可能返回ReplayProcessor实例，因此他将共享最新的缓存消息并在可用时发送新消息
		return messageStream;
	}

    @GetMapping("/getMessageStream")
    public List<MessageResponse> getProductById(){
    	List<MessageResponse> result = new ArrayList<>();
    	result.add(new MessageResponse("abc"));
    	return result;
    }

    @GetMapping("/getMessageAfter")
    public List<MessageResponse> delete(){
    	List<MessageResponse> result = new ArrayList<>();
    	result.add(new MessageResponse("ddd"));
    	return result;
    }

}